#include "zrpc/net/tcp_connection.h"

#include "zrpc/base/logger.h"
#include "zrpc/net/event_loop.h"
#include "zrpc/net/socket.h"

namespace zrpc {
#define MIN_MSG_SIZE (sizeof(Message))
#define INIT_READ_BUFF_SIZE 65536

TcpConnection::TcpConnection(EventLoop *loop, int32_t sockfd,
                             const std::any &context)
    : loop_(loop),
      sockfd_(sockfd),
      reading_(true),
      state_(kConnecting),
      channel_(new Channel(loop_, sockfd_)),
      context_(context) /*,
       rmsg_{nullptr},
       rcap_(0),
       rpos_(0) */
{
  channel_->SetReadCallback(std::bind(&TcpConnection::HandleRead, this));
  channel_->SetWriteCallback(std::bind(&TcpConnection::HandleWrite, this));
  channel_->SetCloseCallback(std::bind(&TcpConnection::HandleClose, this));
  channel_->SetErrorCallback(std::bind(&TcpConnection::HandleError, this));

  // rmsg_.header = static_cast<Message *>(malloc(INIT_READ_BUFF_SIZE));
  // rcap_ = INIT_READ_BUFF_SIZE;
}

TcpConnection::~TcpConnection() {
  assert(state_ == kDisconnected);
  socket::Close(sockfd_);

  // free(rmsg_.header);
  // rmsg_.header = nullptr;
  // rpos_ = 0;
  // rcap_ = 0;
}

void TcpConnection::Shutdown() {
  if (state_ == kConnected) {
    SetState(kDisconnecting);
    loop_->RunInLoop(std::bind(&TcpConnection::ShutdownInLoop, this));
  }
}

void TcpConnection::SetState(StateE s) { state_ = s; }

void TcpConnection::ForceClose() {
  if (state_ == kConnected || state_ == kDisconnecting) {
    SetState(kDisconnecting);
    loop_->QueueInLoop(
        std::bind(&TcpConnection::ForceCloseInLoop, shared_from_this()));
  }
}

void TcpConnection::ForceCloseInLoop() {
  loop_->AssertInLoopThread();
  if (state_ == kConnected || state_ == kDisconnecting) {
    // as if we received 0 byte in HandleRead();
    HandleClose();
  }
}

void TcpConnection::ShutdownInLoop() {
  loop_->AssertInLoopThread();
  if (!channel_->IsWriting()) {
    if (socket::Shutdown(sockfd_) < 0) {
    }
  }
}

// bool TcpConnection::ParseMsgBuffer() {
//  while (rpos_ >= MIN_MSG_SIZE) {
//    uint16_t msg_size = rmsg_.header->size;
//    if (msg_size < MIN_MSG_SIZE) {
//      // message protocol error
//      return false;
//    }
//
//    if (rpos_ < msg_size) {
//      break;
//    }
//
//    message1_callback_(shared_from_this(), *rmsg_.header);
//
//    memmove(rmsg_.buffer, rmsg_.buffer + msg_size, rpos_ - msg_size);
//    rpos_ -= msg_size;
//  }
//
//  return true;
//}
//
// void TcpConnection::AdjustRevBuffer(uint16_t new_size) {
//  if (new_size <= rcap_) {
//    return;
//  }
//
//  Message *new_rmsg = static_cast<Message *>(malloc(new_size));
//  rcap_ = new_size;
//  memmove(new_rmsg, rmsg_.buffer, rpos_);
//
//  free(rmsg_.header);
//  rmsg_.header = new_rmsg;
//}
//
// ssize_t TcpConnection::Read(int32_t fd, int32_t *save_errno) {
//  ssize_t n = socket::Read(channel_->Getfd(), rmsg_.buffer + rpos_,
//                           (rcap_ - rpos_) > 0 ? (rcap_ - rpos_) : 1);
//  if (n < 0) {
//    *save_errno = errno;
//  } else {
//    rpos_ += n;
//  }
//  return n;
//}

void TcpConnection::HandleRead() {
  loop_->AssertInLoopThread();
  int save_errno = 0;
  ssize_t n;

  n = intput_buffer_.ReadFd(channel_->Getfd(), &save_errno);
  if (n > 0) {
    message_callback_(shared_from_this(), &intput_buffer_);
  }
#ifdef _WIN64
  else if (n == 0 || save_errno == WSAECONNRESET) {
    HandleClose();
  } else {
    errno = save_errno;
    HandleError();
  }
#else
  else if (n == 0) {
    HandleClose();
  } else {
    errno = save_errno;
  }
#endif
}

void TcpConnection::HandleWrite() {
  loop_->AssertInLoopThread();

  if (channel_->IsWriting()) {
    ssize_t n = socket::Write(channel_->Getfd(), output_buffer_.Peek(),
                              output_buffer_.ReadableBytes());
    if (n > 0) {
      output_buffer_.Retrieve(n);
      if (output_buffer_.ReadableBytes() == 0) {
        channel_->DisableWriting();
        if (write_complete_callback_) {
          loop_->QueueInLoop(
              std::bind(write_complete_callback_, shared_from_this()));
        }

        if (state_ == kDisconnecting) {
          ShutdownInLoop();
        }
      }
    } else {
    }
  } else {
  }
}

void TcpConnection::HandleClose() {
  loop_->AssertInLoopThread();
  assert(state_ == kConnected || state_ == kDisconnecting);
  SetState(kDisconnected);
  channel_->DisableAll();
  std::shared_ptr<TcpConnection> guard_this(shared_from_this());
  connection_callback_(guard_this);
  close_callback_(guard_this);
}

void TcpConnection::StartRead() {
  loop_->RunInLoop(std::bind(&TcpConnection::StartReadInLoop, this));
}

void TcpConnection::StopRead() {
  loop_->RunInLoop(std::bind(&TcpConnection::StopReadInLoop, this));
}

void TcpConnection::StartReadInLoop() {
  loop_->AssertInLoopThread();
  if (!reading_ || !channel_->IsReading()) {
    channel_->EnableReading();
    reading_ = true;
  }
}

void TcpConnection::StopReadInLoop() {
  loop_->AssertInLoopThread();
  if (reading_ || channel_->IsReading()) {
    channel_->DisableReading();
    reading_ = false;
  }
}

void TcpConnection::ForceCloseDelay() { ForceClose(); }

void TcpConnection::ForceCloseWithDelay(double seconds) {
  if (state_ == kConnected || state_ == kDisconnecting) {
    SetState(kDisconnecting);
    loop_->RunAfter(
        seconds, false,
        std::bind(&TcpConnection::ForceCloseDelay, shared_from_this()));
  }
}

void TcpConnection::HandleError() {
  int err = socket::GetSocketError(channel_->Getfd());
  LOG_SYSERR << "TcpConnection::HandleError ["
             << "] - SO_ERROR = " << err << " " << strerror(err);
}

void TcpConnection::SendInLoopPipe() {
  if (state_ == kConnected) {
    if (loop_->IsInLoopThread()) {
      SendPipe();
    } else {
      void (TcpConnection::*fp)() = &TcpConnection::SendPipe;
      loop_->RunInLoop(std::bind(fp, this));
    }
  }
}

void TcpConnection::SendPipe() {
  loop_->AssertInLoopThread();
  if (!channel_->IsNoneEvent()) {
    if (!channel_->IsWriting()) {
      channel_->EnableWriting();
    }
  }
}

void TcpConnection::SendPipe(Buffer *buf) {
  if (state_ == kConnected) {
    if (loop_->IsInLoopThread()) {
      SendPipeInLoop(buf->Peek(), buf->ReadableBytes());
    } else {
      void (TcpConnection::*fp)(const std::string_view &message) =
          &TcpConnection::SendPipeInLoop;
      loop_->RunInLoop(std::bind(fp, this, buf->RetrieveAllAsString()));
    }
  }
}

void TcpConnection::SendPipe(const void *message, int len) {
  SendPipe(std::string_view(static_cast<const char *>(message), len));
}

void TcpConnection::SendPipe(const std::string_view &message) {
  if (state_ == kConnected) {
    if (loop_->IsInLoopThread()) {
      SendPipeInLoop(message.data(), message.size());
    } else {
      void (TcpConnection::*fp)(const std::string_view &message) =
          &TcpConnection::SendPipeInLoop;
      loop_->RunInLoop(std::bind(fp, this, std::string(message)));
    }
  }
}

void TcpConnection::Send(const void *message, int len) {
  Send(std::string_view(static_cast<const char *>(message), len));
}

void TcpConnection::Send(const std::string_view &message) {
  if (state_ == kConnected) {
    if (loop_->IsInLoopThread()) {
      SendInLoop(message.data(), message.size());
    } else {
      void (TcpConnection::*fp)(const std::string_view &message) =
          &TcpConnection::SendInLoop;
      loop_->RunInLoop(std::bind(fp, this, std::string(message)));
      // loop_->RunInLoop(std::bind(&bindSendInLoop,this,std::string(message)));
    }
  }
}

void TcpConnection::SendPipeInLoop(const std::string_view &message) {
  SendPipeInLoop(message.data(), message.size());
}

void TcpConnection::SendPipeInLoop(const void *message, size_t len) {
  output_buffer_.Append(message, len);
  if (!channel_->IsWriting()) {
    channel_->EnableWriting();
  }
}

void TcpConnection::BindSendPipeInLoop(TcpConnection *conn,
                                       const std::string_view &message) {
  conn->SendPipeInLoop(message.data(), message.size());
}

void TcpConnection::BindSendInLoop(TcpConnection *conn,
                                   const std::string_view &message) {
  conn->SendInLoop(message.data(), message.size());
}

void TcpConnection::Send(Buffer *buf) {
  if (state_ == kConnected) {
    if (loop_->IsInLoopThread()) {
      SendInLoop(buf->Peek(), buf->ReadableBytes());
      buf->RetrieveAll();
    } else {
      loop_->RunInLoop(
          std::bind(&BindSendInLoop, this, buf->RetrieveAllAsString()));
    }
  }
}

void TcpConnection::SendInLoop(const std::string_view &message) {
  SendInLoop(message.data(), message.size());
}

void TcpConnection::SendInLoop(const void *data, size_t len) {
  loop_->AssertInLoopThread();
  ssize_t nwrote = 0;
  size_t remaining = len;
  bool fault_error = false;

  if (state_ == kDisconnected) {
    return;
  }

  if (!channel_->IsWriting() && output_buffer_.ReadableBytes() == 0) {
    nwrote = socket::Write(channel_->Getfd(), data, len);
    if (nwrote >= 0) {
      remaining = len - nwrote;
      if (remaining == 0 && write_complete_callback_) {
        loop_->QueueInLoop(
            std::bind(write_complete_callback_, shared_from_this()));
      }
    } else  // nwrote < 0
    {
      nwrote = 0;
      if (errno != EWOULDBLOCK) {
        if (errno == EPIPE || errno == ECONNRESET)  // FIXME: any others?
        {
          fault_error = true;
        }
      }
    }
  }

  assert(remaining <= len);
  if (!fault_error && remaining > 0) {
    size_t old_len = output_buffer_.ReadableBytes();
    if (old_len + remaining >= high_water_mark_ && old_len < high_water_mark_ &&
        high_water_mark_callback_) {
      loop_->QueueInLoop(std::bind(high_water_mark_callback_,
                                   shared_from_this(), old_len + remaining));
    }

    output_buffer_.Append(static_cast<const char *>(data) + nwrote, remaining);
    if (!channel_->IsWriting()) {
      channel_->EnableWriting();
    }
  }
}

void TcpConnection::ConnectEstablished() {
  loop_->AssertInLoopThread();
  assert(state_ == kConnecting);
  SetState(kConnected);
  channel_->SetTie(shared_from_this());
  channel_->EnableReading();
  connection_callback_(shared_from_this());
}

void TcpConnection::ConnectDestroyed() {
  loop_->AssertInLoopThread();
  if (state_ == kConnected) {
    SetState(kDisconnected);
    channel_->DisableAll();
    connection_callback_(shared_from_this());
  }
  channel_->Remove();
}

}  // namespace zrpc